#include "um_basic_types.h"
#include "umapi_libc.h"
#include "umapi_os.h"
#include "httpTools.h"
#include <sys/epoll.h>
#include "errno.h"
static UM_HANDLE m_report_mutex, m_report_cond, epfd;
static UM_S32 wI = 0, rI =0,  count = 0;
#define M_MAX_REPORT 1024
static UM_VOID * mReportList[M_MAX_REPORT];

static UM_VOID connectReport(S_HttpMsg *msg)
{
	UM_HANDLE fd =  umapi_socket_open(UMAPI_AF_INET, UMAPI_SOCK_STREAM, 0);
	UM_S32 val = 1;
	UMAPI_SOCKET_ADDR ipaddr;
	umapi_socket_ioctl(fd, UMAPI_FIONBIO, &val);
	ipaddr.addr.ipv4 = umapi_socket_inet_addr(msg->host);
	ipaddr.sin_port = msg->port;
	ipaddr.sa_family = UMAPI_AF_INET;
	for (;;)
	{
		int ret = umapi_socket_connect(fd, &ipaddr);
		if (ret == 0)
		{
			umapi_free(msg);
			printf("connect Server is Fail 0\n");
			return ;
		}
		else if (ret < 0) 
		{
			if (errno == EINTR)
			{
				continue;
			}
			else if (errno == EINPROGRESS)
			{	
				break;
			}else
			{
				printf("connect Server is Fail 1\n");
				umapi_free(msg);
				return ;
			}
		}
		break;
	}
	msg->p = fd;
	struct epoll_event ep_event;
	ep_event.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;		
	ep_event.data.ptr = msg;
	if(epoll_ctl(epfd, EPOLL_CTL_ADD, (UM_S32)(fd - 1), &ep_event) == -1)
	{
		umapi_socket_close(fd);
		umapi_free(msg);
		return ;
	}
	return;
}
static UM_VOID reportFun(UM_VOID *param)
{
	S_HttpMsg *nMsg = UM_NULL;
	while(1)
	{	
		umapi_os_mutex_lock(m_report_mutex, -1);
		while (count == 0) {
			umapi_os_cond_timewait(m_report_cond, m_report_mutex, -1);
		}
		nMsg = (S_HttpMsg*)mReportList[rI];
		count--;
		rI = (rI + 1) % M_MAX_REPORT;
		umapi_os_mutex_unlock(m_report_mutex);
		connectReport(nMsg);
	}
}
static UM_VOID sendReportMsg(S_HttpMsg *msg)
{
	UM_HANDLE s = msg->p;
	UM_CHAR buff[4096];
	umapi_snprintf(buff, sizeof(buff),"%s %s HTTP/1.1\r\n"
		"Host: %s:%d\r\n"
		"Content-Type: application/json;charset=UTF-8\r\n"
		"Content-Length: %d\r\n"
		"Connection: Close\r\n"
		"\r\n"
		"%s",
		msg->method == 0 ? "GET" : "POST",
		msg->path, msg->host, msg->port, 
		umapi_strlen(msg->body), msg->body);
	umapi_socket_send(s, buff, umapi_strlen(buff), 0);
}
static UM_S32 recvReportMsg(S_HttpMsg *msg)
{
	UM_CHAR buff[1024];
	UM_HANDLE s = msg->p;
	if(umapi_socket_recv(s, buff, 1024, 0)<=0)
	{
		return -1;
	}
	return 0;	
}
static UM_S32 removeSocket(UM_HANDLE fd)
{
	struct epoll_event ep_event;
	ep_event.events = EPOLLOUT | EPOLLET;	
	ep_event.data.fd = fd -1 ;
	if(epoll_ctl(epfd, EPOLL_CTL_DEL, (UM_S32)(fd - 1), &ep_event) == -1)
	{
		 
	}
	return UM_SUCCESS;
}
static UM_VOID createWriteEpollTask(UM_VOID *param)
{
	int cpunum = get_nprocs();
	umapi_os_mutex_create(UM_NULL,UM_OS_MUTEX_ATTR_RECURSIVE_NP,&m_report_mutex);
	umapi_os_cond_create(UM_NULL,&m_report_cond);
	epfd = epoll_create(cpunum * M_MAX_REPORT);	
	*((UM_S32*)param) = 1;
	for(int i = 0; i<cpunum - 1; i++)
	{
		UM_HANDLE taskHandle;
		umapi_os_task_create(UM_NULL, reportFun, UM_NULL, 8 ,0 , &taskHandle);
	}
	UM_S32 epollnum = cpunum * M_MAX_REPORT;
	
	struct epoll_event *ep_event_arr = (struct epoll_event *)umapi_malloc(sizeof(struct epoll_event) * epollnum);
	while(1)
	{
		UM_S32 nfds = epoll_wait(epfd, ep_event_arr, epollnum, -1);
		if(nfds <=0)
			umapi_os_task_sleep(10);
		for(int i =0; i<nfds; i++)
		{
			struct epoll_event * p = &ep_event_arr[i];
			S_HttpMsg *msg  = (S_HttpMsg *)p->data.ptr;
			if(!msg)
				continue;
			if(p->events & EPOLLRDHUP)
			{
				umapi_socket_close((UM_S32)msg->p);
				removeSocket((UM_S32)msg->p);
				umapi_free(msg);
				p->data.ptr = UM_NULL; 
			}
			else 
			{
				if(p->events & EPOLLOUT)
				{
					sendReportMsg(msg);
				}
				if(p->events & EPOLLIN)
				{
					if(recvReportMsg(msg) < 0)
					{
						umapi_socket_close((UM_S32)msg->p);
						removeSocket((UM_S32)msg->p);
						umapi_free(msg);
						p->data.ptr = UM_NULL;
					}
				}
			}
		}
	}
}
UM_VOID loopReadReportCmd(int fd)
{
	UM_S32 initflag = 0;
	UM_HANDLE taskHandle = UM_NULL;
	umapi_os_task_create(UM_NULL, createWriteEpollTask, &initflag, 8 ,0 , &taskHandle);
	while(!initflag)
	{
		umapi_os_task_sleep(10);
	}
	UM_S32 rlen;
	while(1)
	{
		S_HttpMsg *nMsg = umapi_malloc(sizeof(S_HttpMsg)); 
		rlen = read(fd, nMsg, sizeof(S_HttpMsg));
	//	printf("read new msg msg:{%s:%d %s}\n",
	//		nMsg->host,nMsg->port,nMsg->body); 
		if(rlen == sizeof(S_HttpMsg))
		{
			umapi_os_mutex_lock(m_report_mutex, -1);
			mReportList[wI++] = nMsg;
			wI %= M_MAX_REPORT;
			count++;
			umapi_os_cond_signal(m_report_cond);
			umapi_os_mutex_unlock(m_report_mutex);
		}
	}
}

